package com.codesuc.aichat.aiAgent.strategy.impl;

import cn.dev33.satoken.stp.StpUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.dashscope.aigc.multimodalconversation.AudioParameters;
import com.alibaba.dashscope.audio.asr.recognition.Recognition;
import com.alibaba.dashscope.audio.asr.recognition.RecognitionParam;
import com.alibaba.dashscope.audio.qwen_tts_realtime.*;
import com.alibaba.dashscope.exception.NoApiKeyException;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.codesuc.aichat.aiAgent.config.AIConfig;
import com.codesuc.aichat.aiAgent.config.RealtimePcmPlayer;
import com.codesuc.aichat.aiAgent.constant.AIAgentConstant;
import com.codesuc.aichat.aiAgent.enums.RoleVoiceEnum;
import com.codesuc.aichat.aiAgent.request.ChatRequest;
import com.codesuc.aichat.aiAgent.request.Message;
import com.codesuc.aichat.aiAgent.response.ChatResponse;
import com.codesuc.aichat.aiAgent.strategy.AIService;
import com.codesuc.aichat.chatHistory.entity.ChatHistory;
import com.codesuc.aichat.chatHistory.enums.RoleTypeEnum;
import com.codesuc.aichat.chatHistory.service.ChatHistoryService;
import com.codesuc.aichat.constant.CommonConstant;
import com.codesuc.aichat.enums.CodeSucErrorEnum;
import com.codesuc.aichat.exception.CodeSucException;
import com.codesuc.aichat.role.entity.Role;
import com.codesuc.aichat.role.service.RoleService;
import com.google.gson.JsonObject;
import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;

import javax.sound.sampled.AudioFormat;
import javax.sound.sampled.AudioSystem;
import javax.sound.sampled.LineUnavailableException;
import javax.sound.sampled.TargetDataLine;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicReference;

/**
 * @Author: ZhangGongMing
 * @CreateTime: 2025/9/25 23:21
 * @description: 阿里通义 AI 实现类
 * @Version: 1.0
 */
@Slf4j
@Service
public class TongyiAIServiceImpl implements AIService {

    @Autowired
    private AIConfig config;
    @Autowired
    private ChatHistoryService chatHistoryService;
    @Autowired
    private RoleService roleService;

    private static volatile boolean isRecording = true;

    /**
     * 获取 AI 服务商名称
     *
     * @return AI 服务商名称
     */
    @Override
    public String getServiceName() {
        return AIAgentConstant.TONGYI;
    }

    /**
     * 聊天
     *
     * @param request 聊天请求参数
     * @return 聊天响应结果
     */
    @Override
    public Flux<ChatResponse> chat(ChatRequest request) {
        AIConfig.TongYi tongyi = config.getTongyi();
        if (tongyi == null) {
            throw new CodeSucException(CodeSucErrorEnum.NOT_FOUND_AI_SERVICE_CONFIGURATION);
        }

        // 获取配置项
        String apiKey = tongyi.getApiKey();
        String chatUrl = tongyi.getChatUrl();
        String model = request.getModel();
        if (StrUtil.isBlank(model)) {
            // 如果传了用用户选择的，没有传用默认的
            model = tongyi.getChatModel();
        }

        List<Message> messageList = new ArrayList<>(request.getMessageList());

        boolean isLogin = StpUtil.isLogin();
        long userId = isLogin ? StpUtil.getLoginIdAsLong() : 0L;
        if (isLogin) {
            // 保存当前用户问题
            ChatHistory nowChat = new ChatHistory();
            nowChat.setContent(messageList.getLast().getContent());
            nowChat.setRoleId(request.getRoleId());
            nowChat.setUserId(userId);
            nowChat.setIsUser(RoleTypeEnum.USER.getCode());
            chatHistoryService.save(nowChat);
        }

        // 防止 messageList 太长
        this.checkMessageLength(messageList);

        if (request.getRoleId() != null) {
            // 加入角色设定提示词
            Role role = roleService.getById(request.getRoleId());
            if (role != null) {
                Message message = new Message(RoleTypeEnum.ASSISTANT.getDesc(), role.getPrompt());
                messageList.addFirst(message);
            } else {
                messageList.addFirst(new Message(RoleTypeEnum.ASSISTANT.getDesc(), CommonConstant.AI_CHAT_DEFAULT_PROMPT));
            }
        }

        // 构建请求体
        Map<String, Object> params = new HashMap<>();
        params.put("model", model);
        params.put("messages", messageList);
        params.put("stream", request.getIsStream());

        WebClient webClient = WebClient.builder()
                .baseUrl(chatUrl)
                .defaultHeader("Authorization", "Bearer " + apiKey)
                .defaultHeader("Content-Type", "application/json")
                .build();

        StringBuffer sb = new StringBuffer();
        return Flux.create(sink -> {
            webClient.post()
                    .uri(chatUrl)
                    .bodyValue(params)
                    .retrieve()
                    .bodyToFlux(String.class)
                    .subscribe(data -> {
                                try {
                                    // 处理流式响应（SSE）
                                    if (request.getIsStream()) {
                                        if (!"[DONE]".equals(data)) {
                                            JSONObject jsonObject = JSONObject.parseObject(data);
                                            if (jsonObject.containsKey("choices")) {
                                                JSONObject choice = jsonObject.getJSONArray("choices").getJSONObject(0);
                                                String delta = choice.getJSONObject("delta").getString("content");
                                                sb.append(delta);
                                                log.info("stream response : {}", data);
                                                ChatResponse chatResponse = new ChatResponse();
                                                chatResponse.setSuccess(true);
                                                chatResponse.setContent(delta);
                                                sink.next(chatResponse);
                                            }
                                        } else {
                                            // 收到[DONE]标记，完成流
                                            sink.complete();
                                        }
                                    } else {
                                        // 处理非流式响应（完整JSON）
                                        JSONObject jsonObject = JSONObject.parseObject(data);
                                        if (jsonObject.containsKey("choices")) {
                                            JSONArray choices = jsonObject.getJSONArray("choices");
                                            if (choices.size() > 0) {
                                                JSONObject choice = choices.getJSONObject(0);
                                                JSONObject message = choice.getJSONObject("message");
                                                String content = message.getString("content");
                                                sb.append(content);
                                                log.info("non-stream response : {}", data);
                                                ChatResponse chatResponse = new ChatResponse();
                                                chatResponse.setSuccess(true);
                                                chatResponse.setContent(content);
                                                sink.next(chatResponse);
                                            }
                                        }
                                        // 完成流
                                        sink.complete();
                                    }
                                } catch (Exception e) {
                                    log.error("处理响应时发生错误: {}", e.getMessage(), e);
                                    sink.error(e);
                                }
                            },
                            error -> {
                                // 处理错误
                                log.error("处理流时发生错误: {}", error.getMessage(), error);
                                sink.error(error);
                            },
                            () -> {
                                // 添加历史记录到表中
                                if (isLogin) {
                                    ChatHistory chatHistory = new ChatHistory();
                                    chatHistory.setContent(sb.toString());
                                    chatHistory.setRoleId(request.getRoleId());
                                    chatHistory.setUserId(userId);
                                    chatHistory.setIsUser(RoleTypeEnum.ASSISTANT.getCode());
                                    chatHistoryService.save(chatHistory);
                                }

                                // 正常完成流
                                sink.complete();
                            });
        });
    }

    /**
     * 校验历史消息长度，防止超过 Token 限制
     *
     * @param messageList 历史消息列表
     */
    private void checkMessageLength(List<Message> messageList) {
        if (messageList.size() > 30) {
            messageList.removeLast();
        }
    }

    /**
     * 音频转文字
     *
     * @return 文本结果
     */
    @Override
    public String audioASR() {
        // 重新设置录音状态
        isRecording = true;
        // 创建一个Flowable<ByteBuffer>
        Flowable<ByteBuffer> audioSource =
                Flowable.create(
                        emitter -> {
                            new Thread(
                                    () -> {
                                        TargetDataLine targetDataLine = null;
                                        try {
                                            // 创建音频格式
                                            AudioFormat audioFormat = new AudioFormat(16000, 16, 1, true, false);
                                            // 根据格式匹配默认录音设备
                                            targetDataLine = AudioSystem.getTargetDataLine(audioFormat);
                                            targetDataLine.open(audioFormat);
                                            // 开始录音
                                            targetDataLine.start();
                                            ByteBuffer buffer = ByteBuffer.allocate(1024);
                                            long start = System.currentTimeMillis();
                                            // 录音30s并进行实时转写
                                            while (isRecording && System.currentTimeMillis() - start < 30000) {
                                                int read = targetDataLine.read(buffer.array(), 0, buffer.capacity());
                                                if (read > 0) {
                                                    buffer.limit(read);
                                                    // 将录音音频数据发送给流式识别服务
                                                    emitter.onNext(buffer);
                                                    buffer = ByteBuffer.allocate(1024);
                                                    // 录音速率有限，防止cpu占用过高
                                                    Thread.sleep(20);
                                                }
                                            }
                                            log.info("ASR Service Stop，录音结束");

                                            // 通知结束转写
                                            emitter.onComplete();
                                        } catch (Exception e) {
                                            log.error("录音过程中发生错误: {}", e.getMessage(), e);
                                            emitter.onError(e);
                                        } finally {
                                            // 确保音频设备资源被正确释放
                                            if (targetDataLine != null) {
                                                try {
                                                    if (targetDataLine.isRunning()) {
                                                        targetDataLine.stop();
                                                    }
                                                    if (targetDataLine.isOpen()) {
                                                        targetDataLine.close();
                                                    }
                                                    log.info("音频设备资源已释放");
                                                } catch (Exception closeException) {
                                                    log.error("关闭音频设备时发生错误: {}", closeException.getMessage(), closeException);
                                                }
                                            }
                                        }
                                    })
                                    .start();
                        },
                        BackpressureStrategy.BUFFER);

        AIConfig.TongYi tongyi = config.getTongyi();
        if (tongyi == null) {
            throw new CodeSucException(CodeSucErrorEnum.NOT_FOUND_AI_SERVICE_CONFIGURATION);
        }
        // 创建Recognizer
        Recognition recognizer = new Recognition();
        // 创建RecognitionParam，audioFrames参数中传入上面创建的Flowable<ByteBuffer>
        RecognitionParam param = RecognitionParam.builder()
                .apiKey(tongyi.getApiKey())
                .model(tongyi.getAsrModel())
                .format("pcm")
                .sampleRate(16000)
                .build();

        // 流式调用接口
        StringBuffer sb = new StringBuffer();
        try {
            log.info("ASR Service init finish，启动录音程序");
            recognizer.streamCall(param, audioSource)
                    .blockingForEach(
                            result -> {
                                if (result.isSentenceEnd()) {
                                    sb.append(result.getSentence().getText());
                                    System.out.println("Final Result: " + result.getSentence().getText());
                                }
                            });
        } catch (NoApiKeyException e) {
            log.error("流式 ASR 识别失败");
        }
        return sb.toString();
    }

    /**
     * 停止音频转文字
     */
    @Override
    public void stopAudioASR() {
        isRecording = false;
    }

    /**
     * 语音合成
     *
     * @param text 文本
     */
    @Override
    @Async(value = "virtualThreadPool")  // 使用虚拟线程消费
    public void audioTTS(String text, Long roleId) {
        AIConfig.TongYi tongyi = config.getTongyi();
        if (tongyi == null) {
            throw new CodeSucException(CodeSucErrorEnum.NOT_FOUND_AI_SERVICE_CONFIGURATION);
        }

        QwenTtsRealtimeParam param = QwenTtsRealtimeParam.builder()
                .model(tongyi.getTtsModel())
                .apikey(tongyi.getApiKey())
                .build();
        AtomicReference<CountDownLatch> completeLatch = new AtomicReference<>(new CountDownLatch(1));
        final AtomicReference<QwenTtsRealtime> qwenTtsRef = new AtomicReference<>(null);

        // 创建实时音频播放器实例
        RealtimePcmPlayer audioPlayer;
        try {
            audioPlayer = new RealtimePcmPlayer(24000);
        } catch (LineUnavailableException e) {
            throw new CodeSucException(CodeSucErrorEnum.TTS_SERVICE_ERROR);
        }

        QwenTtsRealtime qwenTtsRealtime = new QwenTtsRealtime(param, new QwenTtsRealtimeCallback() {
            @Override
            public void onOpen() {
                // 连接建立时的处理
            }

            @Override
            public void onEvent(JsonObject message) {
                String type = message.get("type").getAsString();
                switch (type) {
                    case "session.created":
                        // 会话创建时的处理
                        break;
                    case "response.audio.delta":
                        String recvAudioB64 = message.get("delta").getAsString();
                        // 实时播放音频
                        audioPlayer.write(recvAudioB64);
                        break;
                    case "response.done":
                        // 响应完成时的处理
                        break;
                    case "session.finished":
                        // 会话结束时的处理
                        completeLatch.get().countDown();
                    default:
                        break;
                }
            }

            @Override
            public void onClose(int code, String reason) {
                // 连接关闭时的处理
            }
        });
        qwenTtsRef.set(qwenTtsRealtime);
        try {
            qwenTtsRealtime.connect();
        } catch (Exception e) {
            throw new CodeSucException(CodeSucErrorEnum.TTS_SERVICE_ERROR);
        }
        QwenTtsRealtimeConfig config = QwenTtsRealtimeConfig.builder()
                .voice(RoleVoiceEnum.getVoiceByRoleId(roleId))
                .languageType("Chinese")
                .responseFormat(QwenTtsRealtimeAudioFormat.PCM_24000HZ_MONO_16BIT)
                .mode("server_commit")
                .build();
        qwenTtsRealtime.updateSession(config);
        qwenTtsRealtime.appendText(text);

        qwenTtsRealtime.finish();
        try {
            completeLatch.get().await();
            // 等待音频播放完成并关闭播放器
            audioPlayer.waitForComplete();
            audioPlayer.shutdown();
        } catch (InterruptedException e) {
            throw new CodeSucException(CodeSucErrorEnum.TTS_SERVICE_ERROR);
        }


    }
}
